package network

import org.eclipse.paho.client.mqttv3.*
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
import java.nio.charset.Charset


class MqttService {
    /**
     *  A series of basic connectivity tests to validate that basic functions work:
     *  1、Connect
     *  2、Subscribe
     *  3、Publish
     *  4、Unsubscribe
     *  5、Disconnect
     */
    private val client: MqttClient
    private val brokerUrl = "127.0.0.1"
    val port = 1883
    val clientId = "zh6806"
    val username = "root"
    val password = "123456"

    init {
        client = MqttClient("tcp://$brokerUrl:$port", clientId, MemoryPersistence())
        connect()
    }

    fun connect(): Boolean {
        val connOpts = MqttConnectOptions().apply {
            isCleanSession = true
            userName = username
            password = "123456".toCharArray()
            connectionTimeout = 120
            keepAliveInterval = 60
            isAutomaticReconnect = true
        }
        client.connect(connOpts)
        return client.isConnected
    }

    fun subscribe(topic: List<String>) {
        client.setCallback(object: MqttCallbackExtended {
            override fun connectionLost(cause: Throwable?) {
                println("connect success $cause ")
                if(!client.isConnected){
                    client.reconnect()
                }
            }

            override fun messageArrived(topic: String?, message: MqttMessage?) {
                println(topic)
                if(topic == "x/y/z" && message != null){
                    val messageStr = message.payload?.toString(Charset.defaultCharset())
                    val theVal = messageStr?.toInt()?.plus(1000)
                    publish("a/b/c", theVal.toString())
                }
                if(message != null){
                    println(message.id)
                    println(message.qos)
                    println(message.payload?.toString(Charset.defaultCharset()))
                }
            }

            override fun deliveryComplete(token: IMqttDeliveryToken?) {
                println("deliveryComplete messageId={${token?.messageId}}, topics={${token?.topics}}")
            }

            override fun connectComplete(reconnect: Boolean, serverURI: String?) {
                client.subscribe(topic.toTypedArray())
                println("connect success $reconnect $serverURI")
            }
        })
        client.subscribe(topic.toTypedArray())
    }

    fun publish(topic: String, message: String) {
        val msg = MqttMessage(message.toByteArray())
        msg.qos = 1
        client.publish(topic, msg)
    }

    fun unsubscribe(topic: String) {
        client.unsubscribe(topic)
    }

    fun disconnect() {
        client.disconnect()
    }

}

fun main() {
    val mqtt = MqttService()
    val topicList = listOf("eighn/query_/#", "eighn/command_/#", "eighn/test_/#")
//    mqtt.subscribe(topicList)
//    mqtt.publish("a/b/c", "message from kotlin")
    mqtt.subscribe(listOf("x/y/z"))
}